package com.qupeng.demo.kafka.kafkaspringbootconsumer.consumer;

import com.qupeng.demo.kafka.kafkaspringbootconsumer.config.KafkaProperties;
import com.qupeng.demo.kafka.kafkaspringbootconsumer.model.CommonResponse;
import com.qupeng.demo.kafka.kafkaspringbootconsumer.model.ConsumeSuccessResponse;
import com.qupeng.demo.kafka.kafkaspringbootconsumer.model.Product;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.PutMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;

@RestController
@RequestMapping("/messages")
public class KafkaController {

    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    @Autowired
    private ConsumerFactory<Integer, String> consumerFactory;

    @Autowired
    private ConsumerFactory<Integer, Product> consumerFactoryProduct;

    @Autowired
    private KafkaProperties kafkaProperties;

    @PutMapping("")
    public CommonResponse<ConsumeSuccessResponse<Integer, String>> consume() {
        try(Consumer<Integer, String> consumer = consumerFactory.createConsumer(kafkaProperties.getGroupId(), "once-time")) {
            consumer.subscribe(kafkaProperties.getTopics());
            ConsumerRecords<Integer, String> consumerRecords = consumer.poll(Duration.ofMillis(kafkaProperties.getPollingBlockingTimeout()));
            ConsumeSuccessResponse<Integer, String> consumeSuccessResponse = new ConsumeSuccessResponse<>(consumerRecords);
            return new CommonResponse<>("consumer", consumeSuccessResponse);
        }
    }

    @PutMapping("/products")
    public CommonResponse<ConsumeSuccessResponse<Integer, Product>> consumeProducts() {
        try(Consumer<Integer, Product> consumer = consumerFactoryProduct.createConsumer(kafkaProperties.getGroupId(), "once-time")) {
            consumer.subscribe(kafkaProperties.getTopics());
            ConsumerRecords<Integer, Product> consumerRecords = consumer.poll(Duration.ofMillis(kafkaProperties.getPollingBlockingTimeout()));
            ConsumeSuccessResponse<Integer, Product> consumeSuccessResponse = new ConsumeSuccessResponse<>(consumerRecords);
            return new CommonResponse<>("consumer", consumeSuccessResponse);
        }
    }
}
